# +----------------------------------------------------------------------
# | ChatWork智能聊天办公系统
# +----------------------------------------------------------------------
# | 软件声明: 本系统并非自由软件,未经授权任何形式的商业使用均属非法。
# | 版权保护: 任何企业和个人不允许对程序代码以任何形式任何目的复制/分发。
# | 授权要求: 如有商业使用需求,请务必先与版权所有者取得联系并获得正式授权。
# +----------------------------------------------------------------------
# | Author: ChatWork Team <2474369941@qq.com>
# +----------------------------------------------------------------------
import re
import asyncio
import ahocorasick
from typing import Optional, ClassVar, Set
from common.models.sys import SysSensitiveModel


class ValidUtils:
    """ 验证工具 """

    @classmethod
    def is_mobile(cls, value):
        """ 是否为手机号 """
        pattern = r"^1[3-9]\d{9}$"
        return re.match(pattern, value)

    @classmethod
    def is_email(cls, value):
        """ 是否为邮箱号 """
        pattern = r"^[\w\.-]+@[\w\.-]+\.\w+$"
        return re.match(pattern, value)

    @classmethod
    def is_alpha(cls, value):
        """ 是否为纯字母 """
        pattern = r"^[A-Za-z]+$"
        return re.match(pattern, value)

    @classmethod
    def is_alpha_num(cls, value):
        """ 是否为字母和数字 """
        pattern = r"^[A-Za-z0-9]+$"
        return re.match(pattern, value)

    @classmethod
    def is_alpha_dash(cls, value):
        """ 是否为字母和数字,下划线_及破折号- """
        pattern = r"^[A-Za-z0-9\-\_]+$"
        return re.match(pattern, value)

    @classmethod
    def is_chs(cls, value):
        """ 是否为汉字 """
        pattern = r"^[\x{4e00}-\x{9fa5}\x{9fa6}-\x{9fef}\x{3400}-\x{4db5}\x{20000}-\x{2ebe0}]+$/u"
        return re.match(pattern, value)

    @classmethod
    def is_chs_alpha(cls, value):
        """ 是否为汉字、字母 """
        pattern = r"^[\x{4e00}-\x{9fa5}\x{9fa6}-\x{9fef}\x{3400}-\x{4db5}\x{20000}-\x{2ebe0}a-zA-Z]+$/u"
        return re.match(pattern, value)

    @classmethod
    def is_chs_alpha_num(cls, value):
        """ 是否为汉字、字母和数字 """
        pattern = r"^[\x{4e00}-\x{9fa5}\x{9fa6}-\x{9fef}\x{3400}-\x{4db5}\x{20000}-\x{2ebe0}a-zA-Z0-9]+$/u"
        return re.match(pattern, value)

    @classmethod
    def is_chs_dash(cls, value):
        """ 是否为汉字、字母、数字和下划线_及破折号- """
        pattern = r"^[\x{4e00}-\x{9fa5}\x{9fa6}-\x{9fef}\x{3400}-\x{4db5}\x{20000}-\x{2ebe0}a-zA-Z0-9\_\-]+$/u"
        return re.match(pattern, value)

    @classmethod
    def is_id_card(cls, value):
        """ 是否为有效的身份证格式 """
        pattern = r"(^[1-9]\d{5}(18|19|([23]\d))\d{2}((0[1-9])|(10|11|12))(([0-2][1-9])|10|20|30|31)\d{3}[0-9Xx]$)" \
                  r"|(^[1-9]\d{5}\d{2}((0[1-9])|(10|11|12))(([0-2][1-9])|10|20|30|31)\d{3}$)/"
        return re.match(pattern, value)

    @staticmethod
    def is_json(obj: any) -> bool:
        """
        判断给定的字符串是否是有效的JSON对象。

        Args:
            obj (any): 需要被判断的字符串。

        Returns:
            bool: 如果字符串是有效的JSON对象，返回True；否则返回False。

        Author:
            zero
        """
        import json
        try:
            json.loads(obj)
            return True
        except ValueError:
            return False

    @staticmethod
    def is_number(s: any) -> bool:
        """
        判断给定的字符串是否可以转换为数字（整数或浮点数）。

        Args:
            s (any): 需要被判断的字符串。

        Returns:
            bool: 如果字符串可以转换为数字，则返回True；否则返回False。

        Author:
            zero
        """
        try:
            float(s)
            return True
        except ValueError:
            return False

    @staticmethod
    def is_integer(s: any) -> bool:
        try:
            import re
            return bool(re.match(r"^[1-9]\d*$", s))
        except ValueError:
            return False

    @staticmethod
    def is_datetime(date_str: any) -> bool:
        try:
            from datetime import datetime
            datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S")
            return True
        except ValueError:
            return False


class SensitiveUtil:
    """ 敏感词检测工具 """
    acTree: ClassVar[Optional[ahocorasick.Automaton]] = None
    _lock: ClassVar[asyncio.Lock] = asyncio.Lock()

    @classmethod
    async def _run_in_executor(cls, func, *args):
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, func, *args)

    @classmethod
    async def _build_async(cls, words: Set[str]) -> ahocorasick.Automaton:
        """ 在后台线程构建自动机 """
        def _build():
            automaton = ahocorasick.Automaton()
            for word in words:
                automaton.add_word(word, word)
            automaton.make_automaton()
            return automaton

        return await cls._run_in_executor(_build)

    @classmethod
    async def _load_sensitive_words(cls) -> Set[str]:
        """ 异步加载敏感词库 """
        words = set()
        batch_size = 5000  # 每批处理的数量
        query = SysSensitiveModel.filter(scene=1, is_disable=0, is_delete=0)

        offset = 0
        while True:
            batch = await query.offset(offset).limit(batch_size).all()
            if not batch:
                break
            words.update(item.word for item in batch if item.word)
            offset += batch_size

        if not words:
            words.add("cao ni ma de")
        return words

    @classmethod
    async def build_acTree(cls):
        """ 异步懒加载构建自动机 """
        async with cls._lock:
            if not cls.acTree:
                words = await cls._load_sensitive_words()
                cls.acTree = await cls._build_async(words)
        return cls.acTree

    @classmethod
    async def update_words(cls, new_wordlist):
        """ 异步强制更新词库 """
        async with cls._lock:
            cls.acTree = await cls._build_async(new_wordlist)

    @classmethod
    async def clear(cls):
        """ 清空自动机 """
        async with cls._lock:
            cls.acTree = None

    @classmethod
    async def detect_words(cls, text: str) -> Set[str]:
        """
        检测文本中的敏感词

        Args:
            text (str): 要检测的文本

        Returns:
             Set[str]: 敏感词列表

        Author:
            zero
        """
        if not text:
            return set()

        _automaton = await cls.build_acTree()

        def _detect():
            return {word for _, word in _automaton.iter(text)}

        return await cls._run_in_executor(_detect)

    @classmethod
    async def filter_text(cls, text: str, cipher: str = "*") -> str:
        """
        过滤文本中的敏感词

        Args:
            text (str): 要过滤的文本
            cipher (str): 密文符号

        Returns:
             str: 过滤后的文本

        Author:
            zero
        """
        if not text:
            return text

        _cipher = cipher[0] if cipher else "*"
        _automaton = await cls.build_acTree()

        def _filter():
            chars = list(text)
            for end_idx, word in _automaton.iter(text):
                start_idx = end_idx - len(word) + 1
                chars[start_idx:end_idx + 1] = [_cipher] * len(word)
            return "".join(chars)

        return await cls._run_in_executor(_filter)

    @classmethod
    async def check_text(cls, text: str) -> bool:
        """
        检查文本是否包含敏感词

        Args:
            text (str): 要过滤的文本

        Returns:
             bool: True=不包含, 包含则抛异常

        Author:
            zero
        """
        sensitive_words = await cls.detect_words(text)
        if sensitive_words:
            raise ValueError(",".join(sensitive_words))
        return True
